#! /app/php/bin/php
<?php
/**
 * 队列处理分发守护进程
 * - 创建多个队列处理程序进程
 * - 创建队列健康检查程序,如为僵死进程则关闭
 * - 单个进程处理队列相应脚本
 *
 * 需要可执行权限 chmod +x queued.php
 * 
 * @author 张立冰 <zhanglibing@corp.the9.net>
 *  
 * $Id: queued.php 9 2011-02-23 06:56:47Z sbnnest@gmail.com $
 */
define(ROOT, dirname(dirname(dirname(__FILE__))));
set_time_limit(0);
ini_set("memory_limit", "512M");

require_once(ROOT."/lib/base.inc.php");

function sig_handler()
{
	
}

//启动队列
if($argc == 1)
{
	if(file_exists("pids"))
	{
		$pids = file_get_contents("pids");
		$pids = explode(',', substr($pids, 0, strlen($pids)-1));
		foreach($pids as $pid)
			posix_kill(intval($pid), SIGKILL);
	}
	
	$log = getLog('queue/queue');

	if(is_array($queue_server_ip))
	{
		foreach($queue_server_ip as $key=>$queue_item)
		{
			for($i=0; $i< $queue_item['thread']; $i++)
			{
				@popen("/app/php/bin/php " .ROOT . "/daemon/queue/queued.php $key $i &", 'r');
			}
		}
		$log->setData("Queue started...");	
	}
	else
	{
		$log->setData('config error')->write();
		exit();
	}
}

//处理业务逻辑
if($argc == 3)
{
	file_put_contents("pids", posix_getpid() . ",", FILE_APPEND);
	
	$log = getLog('queue/queue_debug');
	
	$mcq = new Memcache();
	$key = intval($argv[1]);
	$server = $queue_server_ip[$key]['server'];
	
	foreach($server as $item)
	{
		$mcq->addServer($item['host'], $item['port']);
	}
	
	while(1)
	{
		$log->setData(date('H:i:s'))->write();
		while($value = $mcq->get($queue_server_ip[$key]['mckey']))
		{
			list($class, $param) = explode(" ", $value);
			$file = ROOT  . "daemon/queue/script/{$class}.class.php";
			if(file_exists($file))
			{
				include_once($file);
				$obj = new $class();
				$obj->exec($param);
			}
			else
			{
				$log->setData(ROOT."/daemon/queue/script/{$class}.class.php not exists")->write();
			}
		}
		sleep(60);
	}
}

?>